package com.rrd.mex.dw.mr.babel;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rrd.mex.dw.mr.babel.parser.*;
import com.rrd.mex.dw.utils.HadoopUtils;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.mexico.BlackListCheckRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

public class CallRecordsMR extends Configured implements Tool {

    public static final String CRT001 = "\001";
    public static final Pattern CRT001_PATTERN = Pattern.compile("\001");
    private static final List<ParseBase> ParserList = getParseList();

    public static class CallMap extends Mapper<LongWritable, Text, Text, Text> {
        private Set<String> appSet = null;
        private Map<String, List<ParseBase>> parserListMap = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            String appType = context.getConfiguration().get("etl.param.app.type", "");
            if (appType != null && !appType.equals("")) {
                appSet = new HashSet<>(Arrays.asList(appType.split(",")));
            }
            // parser list
            setParserMap();

        }

        private void setParserMap() {
            for (ParseBase parser : getParseList()) {
                List<ParseBase> parsers;
                if (parserListMap.containsKey(parser.getType())) {
                    parsers = parserListMap.get(parser.getType());
                } else {
                    parsers = new ArrayList<>();
                }
                parsers.add(parser);
                parserListMap.put(parser.getType(), parsers);
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            if (value == null || value.toString().equals("")) {
                return;
            }
            String is_data="false";
            try {
                JSONObject jsonObj = JSON.parseObject(value.toString());
                String type = jsonObj.getString("type");
                if(jsonObj.getJSONObject("data")!=null && jsonObj.getJSONObject("data").get("type")!=null && (jsonObj.getJSONObject("data").get("type").equals("COMMON_PHONE_DETECTION") || jsonObj.getJSONObject("data").get("type").equals("MULTI_PLATFORM_DETECTION"))){
                    jsonObj = jsonObj.getJSONObject("data");
                    type = jsonObj.getString("type");
                    is_data="true";
                }
                if (type == null || type.equals("CREDIT_UPLOAD")) {
                    return;
                }
                if (appSet != null && !appSet.contains(jsonObj.getString("type"))) {
                    return;
                }
                List<ParseBase> parsers = parserListMap.get(jsonObj.getString("type"));
                if (parsers != null && parsers.size() > 0) {
//                  BaseModel<?> model = (BaseModel<?>) record;
                    for (ParseBase parser : parsers) {
                        BaseModel<?>  model  = JSONObject.parseObject(is_data.equals("true")?jsonObj.toString():value.toString(),parser.getModel().getClass());
                        parseValue(model, parser, context);

                    }

                } else {
                    System.out.println("miss type:|" + jsonObj.getString("type") + "|");
                }
            } catch (Exception e) {
                context.getCounter("error", "ClassCastException").increment(1);
                System.out.println("文本:|" + value.toString() + "|");
                e.printStackTrace();
//				throw e;
            }

        }

        protected void parseValue(BaseModel<?> model, ParseBase parser, Context context) throws IOException, InterruptedException {
            String successful = "1";
            if (model != null) {
                if (parser.parseJob(context, model, successful)) {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType()).increment(1);
                } else {
//                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType() + ".false").increment(1);
                }
            } else {
                context.getCounter("error", "user." + parser.getType() + parser.getSubType() + ".null").increment(1);
            }
        }
    }

    public static class CallRecordReducer extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private NullWritable key2 = NullWritable.get();
        private String dt = null;
        private Map<String, String> map = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
            Configuration conf = context.getConfiguration();
            this.dt = conf.get("etl.param.dt", "2018-04-27");
            this.map = getOutPath(dt);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            this.mos.close();
        }

        private String keystr = null;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            keystr = key.toString().split(",")[0];
            System.out.println("REDUCE KEY:" + keystr + "|" + this.map.get(keystr) + "|" + key.toString());
            for (Text val : values) {
                if (this.map.get(keystr) == null) {
                    System.out.println("ERROR KEY:|" + keystr);
                    context.getCounter("reduce", "reduce.null").increment(1);
//					System.out.println(keystr+"==>"+);
                } else {
                    context.getCounter("reduce", "reduce.success").increment(1);
                    this.mos.write("textinfo", key2, val, this.map.get(keystr));
                }
            }

        }
    }

    private static List<ParseBase> getParseList() {
        ArrayList<ParseBase> parsers = new ArrayList<>();
        parsers.add(new MexPlistRecordParse());
        parsers.add(new MexCalendarRecordParse());
        parsers.add(new MexGpsInfo());
        parsers.add(new MexGpsInfoV2());
        parsers.add(new MexPhoneBookRecordParse());
        parsers.add(new MexSmsReportRecordParse());
        parsers.add(new BlackListCheckRecordParse());
        parsers.add(new CommonPhoneDetectionRecordParse());
        parsers.add(new MultiPlatformDetectionRecordParse());
        parsers.add(new MultiPlatformOverdueRecordParse());
        parsers.add(new PhoneRiskQueryRecordParse());
        parsers.add(new BlacklistReportRecordParse());
        parsers.add(new OverdueReportRecordParse());
        parsers.add(new MultiPlatformReportRecordParse());
        return parsers;
    }

    private static Map<String, String> getOutPath(String dt) {
        Map<String, String> map = new HashMap<>();
        for (ParseBase parser : ParserList) {
            map.put(parser.getType() + parser.getSubType(), "/data/fdm/" + parser.getTableName() + "/dt=" + dt + "/");
        }
        return map;
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
        Job job = Job.getInstance(conf);
        String dt = conf.get("etl.param.dt", "2018-04-19");
        Map<String, String> map = getOutPath(dt);

        System.setProperty("HADOOP_USER_NAME", "hdfs");

        String input = conf.get("etl.param.input", "/data/hbase/call_records/" + dt);
        String output = conf.get("etl.param.output", "/tmp/bdm/call_records");
        String type = conf.get("etl.param.app.type", "");
        String delPaths = conf.get("etl.param.delete.paths");
        if (dt.equals("")) {
            System.out.println("请输入日期：etl.param.dt");
            return -1;
        }
        if (output.equals("")) {
            System.out.println("请输入输出路径：etl.param.output");
            return -1;
        }
        if (!type.equals("")) {
            System.out.println("此次单独输出：" + type);
        }
        System.out.println("pdl输入路径###:" + input);
        System.out.println("输出路径:" + output);
        String isFilterTmp = conf.get("param.is.filter.tmp", "false");
        HadoopUtils.delete(conf, new Path(output));
        List<String> typeList = null;
        if (type != null && !type.equals("")) {
            typeList = Arrays.asList(type.split(","));
        }
        for (String s : map.keySet()) {
            if (typeList != null && typeList.size() > 0) {
                String ss = s.split("\\.")[0];
                if (typeList.contains(ss)) {
                    System.out.println("删除指定目标目录："+map.get(s));
                    HadoopUtils.delete(conf, new Path(map.get(s)));
                }
            } else {
                HadoopUtils.delete(conf, new Path(map.get(s)));
            }
        }
        HadoopUtils.delete(conf, new Path("/tmp/data/userbehavior"));
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        if (isFilterTmp.equals("true")) {
            FileSystem fs = FileSystem.get(new Configuration());
            FileStatus[] status = fs.globStatus(new Path(input + "/*"), new RegexExcludePathFilter(".*\\.data(?<!\\.tmp)$"));
            Path[] listedPaths = FileUtil.stat2Paths(status);
            if (listedPaths == null || listedPaths.length <= 0) {
                System.out.println("输入路径除去.tmp结尾的文件，没有其他文件！");
                System.exit(0);
            }
            TextInputFormat.setInputPaths(job, listedPaths);
        } else {
            TextInputFormat.setInputPaths(job, input);
        }
        TextOutputFormat.setOutputPath(job, new Path(output));
        job.setJarByClass(this.getClass());
        job.setMapperClass(CallMap.class);
        job.setReducerClass(CallRecordReducer.class);
        MultipleOutputs.addNamedOutput(job, "textinfo", TextOutputFormat.class, Text.class, Text.class);

        int stat = job.waitForCompletion(true) ? 0 : 1;
        System.out.println("異常次數：" + job.getCounters().getGroup("error").findCounter("ClassCastException").getValue());
        for (ParseBase parser : ParserList) {
            if (typeList != null && typeList.size() > 0) {
                if (typeList.contains(parser.getType())) {
                    System.out.println(parser.getComment() + parser.getSubType() + ":" + job.getCounters().getGroup("counter").findCounter("user." + parser.getType() + parser.getSubType()).getValue());
                    System.out.println(parser.getComment() + parser.getSubType() + "查询is Null次数：" + job.getCounters().getGroup("error").findCounter("user." + parser.getType() + parser.getSubType() + ".null").getValue());
                }
            }else{
                System.out.println(parser.getComment() + parser.getSubType() + ":" + job.getCounters().getGroup("counter").findCounter("user." + parser.getType() + parser.getSubType()).getValue());
                System.out.println(parser.getComment() + parser.getSubType() + "查询is Null次数：" + job.getCounters().getGroup("error").findCounter("user." + parser.getType() + parser.getSubType() + ".null").getValue());
            }

        }
        System.out.println("reduce.null次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.null").getValue());
        System.out.println("reduce.success次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.success").getValue());

        return stat;
    }

    private class RegexExcludePathFilter implements PathFilter {
        private final String regex;

        public RegexExcludePathFilter(String regex) {
            this.regex = regex;
        }

        @Override
        public boolean accept(Path path) {
            System.out.println("forpath:" + path + "," + path.toString().matches(regex));
            return path.toString().matches(regex);
        }
    }

    public static void main(String[] args) {
        int n = 0;
        try {
            n = ToolRunner.run(new CallRecordsMR(), args);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        System.exit(n);
    }
}
